package com.gmall.realtime.dws.app;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.gmall.realtime.common.base.BaseApp;
import com.gmall.realtime.common.bean.TrafficPageViewBean;
import com.gmall.realtime.common.constant.Constant;
import com.gmall.realtime.common.function.DorisMapFunction;
import com.gmall.realtime.common.util.DateFormatUtil;
import com.gmall.realtime.common.util.FlinkSinkUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.ProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Iterator;

public class DwsTrafficVcChArIsNewPageViewWindow extends BaseApp {
    public static void main(String[] args) {
        new DwsTrafficVcChArIsNewPageViewWindow().start(
                10022,
                1,
                "dws_traffic_vc_ch_ar_is_new_page_view_window",
                Constant.TOPIC_DWD_TRAFFIC_PAGE
        );
    }

    /**
     * 处理数据流的主要方法，包含ETL、状态计算、窗口聚合和结果输出等步骤
     *
     * @param env Flink流执行环境
     * @param dataStreamSource 从Kafka消费的原始数据流
     */
    @Override
    public void handle(StreamExecutionEnvironment env, DataStreamSource<String> dataStreamSource) {
        // 1. ETL数据--->下游需要什么字段
        // 对原始数据进行清洗和转换，过滤掉不符合要求的数据
        SingleOutputStreamOperator<JSONObject> etlStream = etl(dataStreamSource);

        // key 之前添加水印和之后添加之后
        // 2. 添加水位线，并keyby mid
        // 3. 状态编程(转成实体对象)-->求UV和PV
        // 按设备ID(mid)分组，计算UV(独立访客)、PV(页面访问量)等指标
        SingleOutputStreamOperator<TrafficPageViewBean> keyStream = keyBy(etlStream);

        // 为数据流添加水位线，允许3秒的乱序时间
        SingleOutputStreamOperator<TrafficPageViewBean> watermarksStream = getWatermarksStream(keyStream);

        // 4. 进行key ,key的字段vc ch ar is_new
        // 5. reduce累计在一起
        // 按维度(vc:版本, ch:渠道, ar:地区, is_new:是否新用户)分组
        // 在10秒的滚动窗口内聚合指标数据
        SingleOutputStreamOperator<TrafficPageViewBean> reduceStream = getReduce(watermarksStream);

        // 6. 写出到doris api
        // 将聚合结果映射为Doris格式并写入Doris数据库
        reduceStream
                .map(new DorisMapFunction<>())
                .sinkTo(FlinkSinkUtil.getDorisSink(
                        Constant.DORIS_DATABASE + "." + Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW,
                        Constant.DWS_TRAFFIC_VC_CH_AR_IS_NEW_PAGE_VIEW_WINDOW
                ));

    }


    /**
     * 对带有水位线的数据流进行窗口聚合操作
     * 1. 按维度(vc:版本, ch:渠道, ar:地区, is_new:是否新用户)进行分组
     * 2. 使用10秒的滚动事件时间窗口
     * 3. 在窗口内对指标进行累加聚合
     * 4. 添加窗口开始时间、结束时间和当前日期信息
     *
     * @param watermarksStream 已添加水位线的数据流
     * @return 聚合后的数据流，包含各维度组合的页面访问指标
     */
    private SingleOutputStreamOperator<TrafficPageViewBean> getReduce(SingleOutputStreamOperator<TrafficPageViewBean> watermarksStream) {
        return watermarksStream.keyBy(new KeySelector<TrafficPageViewBean, String>() {
                    @Override
                    public String getKey(TrafficPageViewBean trafficPageViewBean) throws Exception {
                        // 组合维度字段作为分组key，格式为"vc:ch:ar:is_new"
                        return trafficPageViewBean.getVc() + ":" + trafficPageViewBean.getCh() + ":" + trafficPageViewBean.getAr() + ":" + trafficPageViewBean.getIsNew();
                    }
                })
                // 定义10秒的滚动事件时间窗口
                .window(TumblingEventTimeWindows.of(Time.seconds(10)))
                // 窗口内聚合函数
                .reduce(new ReduceFunction<TrafficPageViewBean>() {
                    @Override
                    public TrafficPageViewBean reduce(TrafficPageViewBean value1, TrafficPageViewBean value2) throws Exception {
                        // 累加各指标值：PV(页面访问量)、UV(独立访客)、SV(会话数)、停留时长
                        value1.setPvCt(value1.getPvCt() + value2.getPvCt());
                        value1.setUvCt(value1.getUvCt() + value2.getUvCt());
                        value1.setSvCt(value1.getSvCt() + value2.getSvCt());
                        value1.setDurSum(value1.getDurSum() + value2.getDurSum());
                        return value1;
                    }
                }, new WindowFunction<TrafficPageViewBean, TrafficPageViewBean, String, TimeWindow>() {
                    @Override
                    public void apply(String s, TimeWindow timeWindow, Iterable<TrafficPageViewBean> iterable, Collector<TrafficPageViewBean> collector) throws Exception {
                        // 获取窗口起止时间并转换为可读格式
                        long start = timeWindow.getStart();
                        long end = timeWindow.getEnd();

                        // 获取聚合结果
                        Iterator<TrafficPageViewBean> iterator = iterable.iterator();
                        TrafficPageViewBean trafficPageViewBean = iterator.next();

                        // 设置窗口统计时间范围和当前日期
                        trafficPageViewBean.setStt(DateFormatUtil.tsToDateTime(start));
                        trafficPageViewBean.setEdt(DateFormatUtil.tsToDateTime(end));
                        trafficPageViewBean.setCur_date(DateFormatUtil.tsToDate(System.currentTimeMillis()));

                        collector.collect(trafficPageViewBean);
                    }
                });
    }


    /**
     * 为数据流添加水位线和事件时间戳
     * <p>
     * 1. 使用有界无序水位线策略，允许3秒的乱序时间
     * 2. 从TrafficPageViewBean中提取时间戳作为事件时间
     * </p>
     *
     * @param keyStream 已按设备ID(mid)分组并处理后的数据流
     * @return 添加了水位线和事件时间戳的数据流，用于后续的窗口计算
     */
    private SingleOutputStreamOperator<TrafficPageViewBean> getWatermarksStream(SingleOutputStreamOperator<TrafficPageViewBean> keyStream) {
        return keyStream.assignTimestampsAndWatermarks(
                // 创建允许3秒乱序的水位线策略
                WatermarkStrategy.<TrafficPageViewBean>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                        // 指定如何从数据中提取事件时间戳
                        .withTimestampAssigner(new SerializableTimestampAssigner<TrafficPageViewBean>() {
                            @Override
                            public long extractTimestamp(TrafficPageViewBean trafficPageViewBean, long l) {
                                // 使用TrafficPageViewBean中的ts字段作为事件时间
                                return trafficPageViewBean.getTs();
                            }
                        }));
    }

    /**
     * 按设备ID(mid)进行分组并处理，计算UV(独立访客)、SV(会话数)等指标
     * <p>
     * 1. 按设备ID(mid)分组
     * 2. 使用状态编程记录每个设备最后访问日期
     * 3. 计算UV(当天首次访问记为1)、SV(无上一页的访问记为1)
     * 4. 构建TrafficPageViewBean对象，包含各维度信息和指标值
     * </p>
     *
     * @param etlStream 经过ETL处理后的JSON数据流
     * @return 包含各维度信息和指标值的TrafficPageViewBean数据流
     */
    private SingleOutputStreamOperator<TrafficPageViewBean> keyBy(SingleOutputStreamOperator<JSONObject> etlStream){
        return etlStream.keyBy(new KeySelector<JSONObject, String>() {
            @Override
            public String getKey(JSONObject jsonObject) throws Exception {
                // 从common对象中获取设备ID(mid)作为分组key
                return jsonObject.getJSONObject("common").getString("mid");
            }
        }).process(new ProcessFunction<JSONObject, TrafficPageViewBean>() {
            // 状态变量，记录每个设备最后访问的日期
            private ValueState<String> lastDtState;

            @Override
            public void open(Configuration parameters) throws Exception {
                // 初始化状态描述符，设置24小时TTL
                ValueStateDescriptor<String> valueStateDescriptor = new ValueStateDescriptor<>("last_dt", String.class);
                valueStateDescriptor.enableTimeToLive(StateTtlConfig.newBuilder(org.apache.flink.api.common.time.Time.hours(24))
                        // 状态更新策略：在创建和写入时更新TTL
                        .setUpdateType(StateTtlConfig.UpdateType.OnCreateAndWrite).build());

                lastDtState = getRuntimeContext().getState(valueStateDescriptor);
            }

            @Override
            public void processElement(JSONObject jsonObject, ProcessFunction<JSONObject, TrafficPageViewBean>.Context context, Collector<TrafficPageViewBean> collector) throws Exception {
                // 提取时间戳并转换为日期格式
                Long ts = jsonObject.getLong("ts");
                String curDt = DateFormatUtil.tsToDate(ts);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");

                // 获取设备上次访问日期
                String lastValue = lastDtState.value();

                // 初始化UV和SV计数
                long uvCt = 0L;
                long svCt = 0L;

                // 如果当前日期与上次访问日期不同，则认为是新访客(UV=1)
                if (!curDt.equals(lastValue)) {
                    uvCt = 1L;
                    lastDtState.update(curDt);  // 更新最后访问日期状态
                }

                // 如果last_page_id为空，则认为是新会话(SV=1)
                String last_page_id = page.getString("last_page_id");
                if (last_page_id == null) {
                    svCt = 1L;
                }

                // 构建并输出TrafficPageViewBean对象
                collector.collect(TrafficPageViewBean.builder()
                        .ar(common.getString("ar"))  // 地区
                        .ch(common.getString("ch"))  // 渠道
                        .vc(common.getString("vc"))  // 版本
                        .isNew(common.getString("is_new"))  // 是否新用户
                        .uvCt(uvCt)  // 独立访客数
                        .svCt(svCt)  // 会话数
                        .pvCt(1L)    // 页面访问量(每次访问记为1)
                        .durSum(page.getLong("during_time"))  // 停留时长
                        .ts(ts)      // 时间戳
                        .sid(common.getString("sid"))  // 会话ID
                        .build());

                /**
                 * 状态TTL示例说明：
                 *   A设备
                 *   2024.11.19 10:00 -- > 2024.11.20 10:00(到期)      1
                 *
                 *   2024.11.20 9:00 ----> 2024.11.21 9:00             1
                 *   2024.11.20 11:00                                  1
                 */
            }
        });
    }


    /**
     * 对Kafka原始数据进行ETL处理
     * <p>
     * 1. 将JSON字符串解析为JSONObject
     * 2. 检查数据完整性：必须包含common和page对象
     * 3. 验证关键字段：设备ID(mid)和时间戳(ts)不能为空
     * 4. 过滤掉不符合条件的数据
     * </p>
     *
     * @param kafkaSource 从Kafka消费的原始数据流，每条数据为JSON格式字符串
     * @return 经过清洗和验证的JSONObject数据流，包含有效且完整的数据
     */
    private SingleOutputStreamOperator<JSONObject> etl(DataStreamSource<String> kafkaSource) {
        return kafkaSource.flatMap(new FlatMapFunction<String, JSONObject>() {
            @Override
            public void flatMap(String s, Collector<JSONObject> collector) throws Exception {
                // 将JSON字符串解析为JSONObject
                JSONObject jsonObject = JSON.parseObject(s);

                // 获取common和page对象，这两个是必须包含的基础对象
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String ts = jsonObject.getString("ts");

                // 数据完整性检查：必须包含common和page对象
                if (common != null && page != null) {
                    // 获取设备ID(mid)，这是用户唯一标识
                    String mid = common.getString("mid");

                    // 关键字段验证：设备ID和时间戳不能为空
                    if (mid != null && ts != null) {
                        // 收集符合条件的数据
                        collector.collect(jsonObject);
                    }
                }
            }
        });
    }

}